package com.broadsoft.cluster.smartcluster;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.jgroups.Address;
import org.jgroups.ChannelException;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.View;
import org.jgroups.util.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.broadsoft.cluster.smartcluster.context.ExecutionContext;
import com.broadsoft.cluster.smartcluster.message.CallTaskMessage;
import com.broadsoft.cluster.smartcluster.message.ControlMessage;
import com.broadsoft.cluster.smartcluster.message.ExecutionContextAwareCallable;
import com.broadsoft.cluster.smartcluster.message.ExecutionContextAwareRunnable;
import com.broadsoft.cluster.smartcluster.message.RunTaskMessage;
import com.broadsoft.cluster.smartcluster.partition.ClusterPartitionTable;

public class ClusterManager extends DefaultReceiverAdapter implements PartitionService {

    private static final Logger logger = LoggerFactory.getLogger(ClusterManager.class);
    private static final ClusterPartitionTable clusterPartitionTable = new ClusterPartitionTable();
    private volatile Address coordinatorAddress;

    private JChannel channel = null;
    private static final ExecutorService executorService = Executors.newCachedThreadPool();

    /**
     * 
     */
    public Object invoke(int partitionID, Callable task) {
        return null;
    }

    /**
     * 
     */
    public void asyncInvoke(int partitionID, Runnable task) {

    }

    /**
     * 
     * @param clusterMessage
     * @param partitionID
     * @throws Exception
     */
    private void _send(ClusterMessage clusterMessage, int partitionID) throws Exception {
        if (logger.isTraceEnabled()) {
            logger.trace("============>Sending the message at the node address[{}].", channel.getAddress());
        }
        byte[] arrayOfByteOnRequest = Util.objectToByteBuffer(clusterMessage);
        Address srcAddress = channel.getAddress();
        Address destAddress = clusterPartitionTable.getMasterAddressByPartitionID(partitionID);
        Message msg = new Message(destAddress, srcAddress, arrayOfByteOnRequest);
        channel.send(msg);

    }

    public void start() throws ChannelException {
        channel = new JChannel("udp.xml");
        channel.setReceiver(this);
        channel.connect("dzone-demo");

    }

    public void stop() throws ChannelException {
        channel.disconnect();
        channel.close();
    }

    /**
     * 
     * Execute the command
     * 
     * @param msg
     */
    public void receive(Message msg) {
        if (logger.isTraceEnabled()) {
            logger.trace("<============Receive the message at the node address[{}].", channel.getAddress());
        }
        ExecutionContext.createContext();
        try {
            _receive(msg);
        } finally {
            ExecutionContext.releaseContext();
        }

    }

    @SuppressWarnings("rawtypes")
    private void _receive(Message msg) {
        Object payload = SerializableHelper.objectFromByteBuffer(msg.getBuffer());
        if (!(payload instanceof ClusterMessage)) {
            logger.warn("Discard invalid message {}", msg);
        }
        switch (((ClusterMessage) payload).getMessageType()) {
        case ClusterMessage.CALL_TASK_MESSAGE:
            handleCallableMessage((CallTaskMessage) payload);
            break;
        case ClusterMessage.RUN_TASK_MESSAGE:
            handleExecutionMessage((RunTaskMessage) payload);
            break;
        case ClusterMessage.CONTROL_MESSAGE:
            handleControlMessage((ControlMessage) payload);
            break;
        }
    }

    private void handleExecutionMessage(RunTaskMessage message) {
        executorService.execute(new ExecutionContextAwareRunnable(message.getTask()));
    }

    private void handleControlMessage(ControlMessage m) {

    }

    private void handleCallableMessage(CallTaskMessage message) {
        Future<Object> future = executorService.submit(new ExecutionContextAwareCallable(message.getTask()));
        try {
            Object object = future.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }

    /**
     * 
     * FIXME: Rebuild the partition table
     * 
     * @param view
     */
    public void viewAccepted(View view) {
        if (logger.isTraceEnabled()) {
            logger.trace("View is {}.", view);
        }
        clusterPartitionTable.enterPartition(view);
    }

    /**
     * FIXME: Rebuild the partition table
     * 
     * @param address
     */
    public void suspect(Address address) {
        clusterPartitionTable.leavePartition(address);
    }

}
